package com.cl.spark.node;

import com.alibaba.fastjson.JSONObject;
import com.cl.spark.base.BaseSparkNode;
import com.cl.spark.dto.SparkParam;
import com.cl.spark.dto.SparkResult;
import com.cl.spark.enums.SparkNodeEnum;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SparkSession;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;


@Component
public class DataSourceNode extends BaseSparkNode {

    @Autowired
    SparkSession sparkSession;

    @Override
    public SparkResult process(SparkParam sparkParam) {
        JSONObject expression = sparkParam.getNodeExpression();
        String tableName = String.valueOf(expression.get("targetTable"));
        String dbCode = String.valueOf(expression.get("dbCode"));

        String url = expression.getString("url");
        String username = expression.getString("username");
        String password = expression.getString("password");
        String skip = expression.getString("skip");
        String limit = expression.getString("limit");

        String dbTable = "select " + String.join(",", sparkParam.getFieldsAs()) + ",'" + tableName + "' as TABLE_NAME,'" + dbCode + "' as DB_CODE " + " from " + tableName + " limit " + skip + "," + limit;

        System.out.println(dbTable);

        // 从数据库读取数据为DataFrame
        Dataset<Row> dataset = sparkSession.read()
                .format("jdbc")
                .option("url", url)
                .option("dbtable", "( " + dbTable + " ) as temp_table")
                .option("user", username)
                .option("password", password)
                .load();

        return SparkResult.success(dataset);

    }

    @Override
    public SparkNodeEnum getType() {
        return SparkNodeEnum.DATA_SOURCE;
    }


}
